/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.typhosoft.jcommon.observer;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;

/**
 *
 * @author SantonocitoM
 */
public class NotifierExecutor {

    private final HashMap<IClient, List<Object>> tobeNotified = new HashMap<>();
    private final ArrayBlockingQueue<NotifierThread> threadPool = new ArrayBlockingQueue(Runtime.getRuntime().availableProcessors() + 1);
    private final NotifierThread[] pool;

    public NotifierExecutor() {
        pool = new NotifierThread[Runtime.getRuntime().availableProcessors() + 1];
        for (int i = 0; i < pool.length; i++) {
            pool[i] = new NotifierThread(this, i);
            pool[i].start();
        }
    }

    void notifyObserver(IClient obs, List<Object> lst) {

        //TraceLogger.info(this, "to be notified  " + obs);

        synchronized (pool) {
            boolean found = false;
            int availableThreadIdx = -1;
            for (int i = 0; i < pool.length; i++) {
                if (pool[i].getConsumer() == obs) {
                    found = true;
                    break;
                } else if (pool[i].isFree()) {
                    availableThreadIdx = i;
                }
            }
            if (found || availableThreadIdx < 0) {
                if (lst != null) {
                    if (!tobeNotified.containsKey(obs)) {
                        tobeNotified.put(obs, lst);
                    } else {
                        tobeNotified.get(obs).addAll(lst);
                    }
                }
                //TraceLogger.info(this, "thread already running for observer " + obs + ", new queue size " + tobeNotified.get(obs).size());
            } else if (availableThreadIdx >= 0) {
                List listToNotify = null;

                if (!tobeNotified.containsKey(obs)) {
                    listToNotify = lst;
                } else {
                    listToNotify = tobeNotified.remove(obs);
                    if (lst != null) {
                        listToNotify.addAll(lst);
                    }
                }

                if (listToNotify != null && !listToNotify.isEmpty()) {
                    pool[availableThreadIdx].addJob(obs, listToNotify);
                }

            }
        }
    }

    void endJob(NotifierThread thread) {
        synchronized (pool) {
            //TraceLogger.info(this, "map size " + tobeNotified.size());
            IClient[] observer = new IClient[tobeNotified.size()];
            //List l = new ArrayList();
            tobeNotified.keySet().toArray(observer);
            for (IClient o : observer) {
                notifyObserver(o, null);
            }
        }
    }
};
